package com.sf.iisp.kafka.adapter;

import com.sf.kafka.api.produce.ProduceConfig;
import com.sf.kafka.api.produce.ProduceOptionalConfig;
import com.sf.kafka.api.produce.ProducerPool;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

public class KafkaProducerPoolDelegate extends ProducerPool implements IKafkaProducerPoolDelegate {
    private ProduceConfig produceConfig;
    private String topic;

    public KafkaProducerPoolDelegate(ProduceConfig produceConfig) {
        super(produceConfig);
    }

    public KafkaProducerPoolDelegate(ProduceConfig produceConfig, ProduceOptionalConfig optionalConfig) {
        super(produceConfig, optionalConfig);
        this.produceConfig = produceConfig;
        this.topic = subTopic(produceConfig.getTopicTokens());
    }

    @Override
    public void sendString(String msg) {
        super.sendString(this.topic, msg);
    }

    @Override
    public void sendByte(byte[] msg) {
        super.sendBytes(this.topic, msg);
    }

    @Override
    public void sendBatchString(List<String> messages) {
        super.batchSendString(this.topic, messages);
    }

    @Override
    public void sendBatchBytes(List<byte[]> messages) {
        super.batchSendBytes(this.topic, messages);
    }

    private String subTopic(String topicToken) {
        return StringUtils.substringBefore(topicToken, ":");
    }
}
